package pro.fraud_detect;

import lombok.AllArgsConstructor;
import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import pro.fraud_detect.beans.Transaction;
import pro.fraud_detect.config.Config;
import pro.fraud_detect.config.Param;
import pro.fraud_detect.config.Parameters;
import pro.fraud_detect.sources.InputSource;

import java.time.Duration;

/***
 * @author hunter.d
 * @qq 657270652
 * @wx haitao-duan
 * @date 2021/2/22
 **/
@AllArgsConstructor
public class RuleEvaluator {
    private Config config;

    public void run() throws Exception {

        Boolean isLocal = config.get(Parameters.LOCAL_EXECUTION);

        Configuration flinkConf = new Configuration();

        StreamExecutionEnvironment env = isLocal ?
                StreamExecutionEnvironment.createLocalEnvironment(flinkConf) :
                StreamExecutionEnvironment.getExecutionEnvironment(flinkConf);

        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        env.getCheckpointConfig().setCheckpointInterval(config.get(Parameters.CHECKPOINT_INTERVAL));
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(config.get(Parameters.MIN_PAUSE_BETWEEN_CHECKPOINTS));
        env.setRestartStrategy(RestartStrategies.noRestart());

        // 读取数据流
        SourceFunction<String> dataSource = InputSource.createTransactionDataSource(env, "kafka", config);

        WatermarkStrategy<Transaction> transactionWatermarkStrategy = WatermarkStrategy
                .<Transaction>forBoundedOutOfOrderness(Duration.ofMillis(config.get(Parameters.OUT_OF_ORDERNESS)))
                .withTimestampAssigner((event, timestamp) -> event.getEventTime());

        SingleOutputStreamOperator<Transaction> transOperator = env
                .addSource(dataSource)
                .map(line -> InputSource.string2Transaction(line))
                .returns(Transaction.class)
                .assignTimestampsAndWatermarks(transactionWatermarkStrategy);

        transOperator.print();

        // 读取规则流，并广播
        SourceFunction<String> ruleStream = InputSource.createRuleSource(env, "kafka", config);

        env.execute();

    }


}
